package com.gugee.ins.data.collector.media.service.impl;

import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.ins.common.dal.service.blogger.InsBloggerDao;
import com.gugee.ins.common.dal.service.media.InsMediaDao;
import com.gugee.ins.data.collector.media.config.InsMediaConfig;
import com.gugee.ins.data.collector.media.service.InsAdMediaIdentifyService;

import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;


import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ForkJoinPool;

/**
 * InsAdMediaIdentifyService 服务
 * @author Huangxin
 */
@Slf4j
@Service
public class InsAdMediaIdentifyServiceImpl implements InsAdMediaIdentifyService {

    /**
     * 配置
     */
    @Autowired
    InsMediaConfig insMediaConfig;

    /**
     * 线程调度
     */
    @Autowired
    ScheduleTemplate scheduleTemplate;

    /**
     * InsBloggerDao
     */
    @Autowired
    InsBloggerDao insBloggerDao;

    /**
     * InsMediaDao
     */
    @Autowired
    InsMediaDao insMediaDao;

    /**
     * kafkaTemplate
     */
    @Autowired
    KafkaTemplate<Long,InsBloggerMessage> kafkaTemplateInsBlogger;

    /**
     * 线程池
     */
    ForkJoinPool forkJoinPool;

    /**
     * 线程名
     */
    private String workName = "insAdMediaIdentify-worker";

    /**
     * 初始化线程池
     */
    @PostConstruct
    public void init() {
        forkJoinPool = ThreadPoolUtils.forkJoinPool(insMediaConfig.getInsMediaCorePollSize(), workName);
        scheduleTemplate.attach(workName, forkJoinPool);
    }

    /**
     * 处理insMedia收入任务
     * @param insBloggerMessage
     */
    @Override
    public void handleInsAdMediaIdentify(List<InsBloggerMessage> insBloggerMessage) {
        try {
            forkJoinPool.submit(() -> insBloggerMessage.parallelStream().forEach(msg -> {
                try {
                    // 识别 广告帖子
                    identifyMedias(msg);
                } catch (Exception ex) {
                    ex.printStackTrace();
                    log.error("InsAdMedia Identify error. uid:{}, error :{}" ,
                            msg.getUid(),ex.getMessage());
                }
            })).get();
        } catch (Exception ex) {
            log.error("InsAdMedia Identify  thread error.  {}" ,  ex.getMessage());
        }
    }

    /**
     * 获取播主帖子数据并进行识别
     * @param insBloggerMessage
     */
    public void identifyMedias(InsBloggerMessage insBloggerMessage){
        long start = System.currentTimeMillis();
        // 获取该播主需要识别的帖子
        List<String> shortcodes = insMediaDao.findAdMediaIdentifyShortCodeByUid(insBloggerMessage.getUid(),
                start / 1000 - insMediaConfig.getLatestDays() * 24 * 3600);
        if(shortcodes != null && !shortcodes.isEmpty()){
            // 识别是否是广告帖子
            shortcodes.forEach(this::identifyMedias);

            // 发送统计
            sendPriorityStMediaAdMsg(insBloggerMessage);
        }

        if ((System.currentTimeMillis() - start) / 1000 > insMediaConfig.getMediaDebugTimeLimit()) {
            log.info("Identify Identify uid:{} use time:{}s", insBloggerMessage.getUid(),(System.currentTimeMillis() - start) / 1000);
        }
    }

    /**
     * 识别帖子数据
     * @param shortcode
     */
    public void identifyMedias(String shortcode){
        final int[] isAd = {2};
        //获取标记的用户数据
        List<InsBloggerMessage> insBloggers = insBloggerDao.findInsBloggerByShortcode(shortcode);
        if(!insBloggers.isEmpty()){
            List<Long> clUid = new ArrayList<>(insBloggers.size());
            List<Long> updateUid = new ArrayList<>(insBloggers.size());
            // 判断是否是广告帖
            insBloggers.forEach(insBlogger -> {
                if(insBlogger.getId() > 0){
                    if(isAd[0] != 1 && insBlogger.getTypeCode() == 1){
                        isAd[0] = 1;
                    }
                    if(insBlogger.getTypeCode() != 1 && insBlogger.getTypeCode() != 0){
                        updateUid.add(insBlogger.getUid());
                    }
                }else{
                    clUid.add(insBlogger.getUid());
                }
            });

            if(!clUid.isEmpty()){
                // 将标记设为未分析，下轮再判断
                if(isAd[0] != 1){
                    isAd[0] = 0;
                }
                // 发送补采消息
                clUid.forEach(this::sendPriorityCollectBloggerMsg);
            }
            if(!updateUid.isEmpty()){
                // 将标记设为未分析，下轮再判断
                if(isAd[0] != 1){
                    isAd[0] = 0;
                }
                // 发送补采消息
                updateUid.forEach(this::sendUpdateBloggerMsg);
            }
        }

        // 更新结论
        if(isAd[0] != 0){
            insMediaDao.updateMediaIsAd(shortcode,isAd[0]);
        }
    }

    /**
     * 发送博主优先收录的消息
     * @param uid
     */
    public void sendPriorityCollectBloggerMsg(long uid){
        if(uid > 0 && insBloggerDao.checkCollecting(uid,true)){
            //这里需要添加验证
            InsBloggerMessage insBloggerMessage = new InsBloggerMessage();
            insBloggerMessage.setUid(uid);
            insBloggerMessage.setUserName("");
            insBloggerMessage.setPicUrl("");
            insBloggerMessage.setRetryLimitTimes(insMediaConfig.getInsBloggerPriorityRetryTimes());
            insBloggerMessage.setRetryTimes(insMediaConfig.getInsBloggerPriorityRetryTimes());

            kafkaTemplateInsBlogger.send(insMediaConfig.getInsBloggerPriorityCollectOnlineTopic(),insBloggerMessage);
        }
    }

    /**
     * 发送博主优先收录的消息
     * @param uid
     */
    public void sendUpdateBloggerMsg(long uid){
        if(uid > 0){
            //这里需要添加验证
            InsBloggerMessage insBloggerMessage = new InsBloggerMessage();
            insBloggerMessage.setUid(uid);
            insBloggerMessage.setUserName("");
            insBloggerMessage.setPicUrl("");
            insBloggerMessage.setRetryLimitTimes(insMediaConfig.getInsBloggerPriorityRetryTimes());
            insBloggerMessage.setRetryTimes(insMediaConfig.getInsBloggerPriorityRetryTimes());

            kafkaTemplateInsBlogger.send(insMediaConfig.getInsBloggerUpdateOnlineTopic(),insBloggerMessage);
        }
    }

    /**
     * 发送优先统计广告帖子消息
     * @param insBloggerMessage
     */
    public void sendPriorityStMediaAdMsg(InsBloggerMessage insBloggerMessage){
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        insBloggerMessage.setRetryTimes(insMediaConfig.getInsMBloggerMediaAdStPriorityRetryTimes());
        insBloggerMessage.setRetryLimitTimes(insMediaConfig.getInsMBloggerMediaAdStPriorityRetryTimes());
        kafkaTemplateInsBlogger.send(insMediaConfig.getInsMBloggerMediaAdStPriorityTopic(),insBloggerMessage);
    }
}
